package com.tencent.qf.bascic.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.tencent.qf.bascic.config.DataCollectProperties;
import com.tencent.qf.bascic.config.DateReportProperties;
import com.tencent.qf.bascic.controller.req.DataCollectRequest;
import com.tencent.qf.bascic.controller.req.DataReportRequest;
import com.tencent.qf.bascic.service.DataReportService;
import com.tencent.qf.common.constants.DataReportApiNameConstants;
import com.tencent.qf.common.exception.ErrorCode;
import com.tencent.qf.common.exception.ServiceException;
import com.tencent.qf.common.pojo.HttpResult;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.SendResult;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import org.springframework.util.DigestUtils;
import org.springframework.util.ObjectUtils;
import org.springframework.util.concurrent.ListenableFuture;

import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;

/**
 * @author: p_haojiawu@tencent.com
 * @date: 2023-05-10 19:21
 **/

@Component
@Slf4j
public class DataReportServiceImpl implements DataReportService {

    @Resource
    private KafkaTemplate<String,String> kafkaTemplate;

    @Value("${spring.kafka.topic.topic-qf-data-report}")
    private String topicDataReport;

    @Resource
    private DateReportProperties dateReportProperties;

    @Resource
    private DataCollectProperties dataCollectProperties;

    @Override
    public String report(DataReportRequest request) {
        JSONObject apiParams = new JSONObject();
        try {
            apiParams = (JSONObject) JSONObject.parse(request.getApiParams());
        }catch (Exception e){
            log.error("转换json异常，request:" + apiParams.toJSONString() + ",", e);
            return "apiParams json转换异常";
        }
        String apiName = request.getApiName();
        String topic = dateReportProperties.getTopics().get(apiName);
        if(topic == null){
            return "非法apiName";
        }
        if(apiName.equals(DataReportApiNameConstants.JD_GIFT_TRANSFER_MESSAGE)){
            String checkResult = checkSpecialAreaInfo(apiParams);
            if(StringUtils.isNotEmpty(checkResult)){
                return checkResult;
            }
        }
        request.setUniqueKey(DigestUtils.md5DigestAsHex(JSONObject.toJSONString(request).getBytes()));
        // 转义params
        JSONObject jsonObject = (JSONObject) JSON.toJSON(request);
        jsonObject.put("apiParams", apiParams);
        String  message = jsonObject.toJSONString();
        log.info("数据上报到kafka， message: " + message);
        ListenableFuture<SendResult<String, String>> send = kafkaTemplate.send(topic, message);
        try{
            SendResult<String, String> stringStringSendResult = send.get(5L, TimeUnit.SECONDS);
            log.info("上报数据,发送Kafka消息结果：{}", JSON.toJSONString(stringStringSendResult));
            return "";
        }catch (Exception e){
            log.error("上报数据，发送kafka消息失败， error", e);
            throw new ServiceException(ErrorCode.SEND_KAFKA_ERROR);
        }
    }

    /**
     * 数据收集
     * @param request
     * @return
     */
    @Override
    public HttpResult<Boolean> dataCollect(DataCollectRequest request) {
        JSONObject content;
        try {
            content = (JSONObject) JSONObject.parse(request.getContentJson());
        }catch (Exception e){
            log.error("转换json异常，request:{}  error:" + request.getContentJson(), e);
            return HttpResult.error(ErrorCode.BAD_REQUEST.getCode(), "contentJson参数，json格式异常");
        }

        String scene = request.getScene();
        String message = content.toJSONString();

        // 检查topic是否正确
        Boolean checkTopic = this.checkSceneTopic(scene);
        if(!checkTopic){
            log.error("非法scene: " + scene);
            return HttpResult.error(ErrorCode.BAD_REQUEST.getCode(), "非法scene");
        }
        // 发送Kafka
        return dataCollectKafka(request, message);
    }

    /**
     * 发送Kafka
     * @param request
     * @param data
     * @return
     */
    private HttpResult<Boolean> dataCollectKafka(DataCollectRequest request, String data) {
        String topic = dataCollectProperties.getTopics().get(request.getScene());
        Message<String> message = MessageBuilder
                .withPayload(data)
                .setHeader(KafkaHeaders.TOPIC, topic)
                .setHeader(KafkaHeaders.MESSAGE_KEY, request.getKey())
                .setHeader("scene", request.getScene())
                .setHeader("source", request.getSource())
                .setHeader("X-Custom-Header", "Sending Custom Header with Spring Kafka")
                .build();

        log.info("数据收集，发送到kafka，topic:{} message:{} ", topic, data);
        ListenableFuture<SendResult<String, String>> send = kafkaTemplate.send(message);
        try{
            SendResult<String, String> stringStringSendResult = send.get(3L, TimeUnit.SECONDS);
            log.info("数据收集，发送到kafka消息结果：{}", JSON.toJSONString(stringStringSendResult));
            return HttpResult.success(true);
        }catch (Exception e){
            log.error("数据收集，发送到kafka消息失败， error", e);
            throw new ServiceException(ErrorCode.SEND_KAFKA_ERROR);
        }
    }

    /**
     * 检查scene是否正确
     * @param scene
     * @return
     */
    private Boolean checkSceneTopic(String scene) {
        return dataCollectProperties.getTopics().containsKey(scene);
    }

    /**
     * 专区信息入参校验
     * @author Lussion
     * @date 2023/8/24 17:33
     * @param apiParams
     * @return java.lang.String
     */
    private String checkSpecialAreaInfo(JSONObject apiParams){
        if(ObjectUtils.isEmpty(apiParams.get("outTradeNo"))){
            return "outTradeNo不能为空";
        }
        if(ObjectUtils.isEmpty(apiParams.get("unionId"))){
            return "unionId不能为空";
        }
//        if(ObjectUtils.isEmpty(apiParams.get("giftCouponKey"))){
//            return "giftCouponKey不能为空";
//        }
        if(ObjectUtils.isEmpty(apiParams.get("level"))){
            return "level不能为空";
        }
        if(ObjectUtils.isEmpty(apiParams.get("skuId"))){
            return "skuId不能为空";
        }
        if(ObjectUtils.isEmpty(apiParams.get("spuId"))){
            return "spuId不能为空";
        }
        if(ObjectUtils.isEmpty(apiParams.get("goodsId"))){
            return "goodsId不能为空";
        }
//        if(ObjectUtils.isEmpty(apiParams.get("couponId"))){
//            return "couponId不能为空";
//        }
        if(apiParams.get("actExtSwitchType").equals(0)){
            if(ObjectUtils.isEmpty(apiParams.get("actExtInfo"))){
                return "actExtInfo不能为空";
            }
        }
//        if(ObjectUtils.isEmpty(apiParams.get("mergeLink"))){
//            return "mergeLink不能为空";
//        }
//        if(ObjectUtils.isEmpty(apiParams.get("cpsLink"))){
//            return "cpsLink不能为空";
//        }
        if(ObjectUtils.isEmpty(apiParams.get("shortUrl"))){
            return "shortUrl不能为空";
        }
        if(ObjectUtils.isEmpty(apiParams.get("generateUrlRequest"))){
            return "generateUrlRequest不能为空";
        }
        if(ObjectUtils.isEmpty(apiParams.get("generateUrlResponse"))){
            return "generateUrlResponse不能为空";
        }
        return "";
    }
}
